package com.xx.rpc.core.remoting.invoker.call;

import com.xx.rpc.core.remoting.net.params.XxRpcFutureResponse;
import com.xx.rpc.core.remoting.net.params.XxRpcResponse;
import com.xx.rpc.core.util.XxRpcException;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 消费者异步响应类
 *
 * @author hanxiyue
 */
public class XxRpcInvokeFuture implements Future {

    //rpc异步响应
    private XxRpcFutureResponse futureResponse;

    public XxRpcInvokeFuture(XxRpcFutureResponse futureResponse) {
        this.futureResponse = futureResponse;
    }
    public void stop(){
        // remove-InvokerFuture
        futureResponse.removeInvokerFuture();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return futureResponse.cancel(mayInterruptIfRunning);
    }

    @Override
    public boolean isCancelled() {
        return futureResponse.isCancelled();
    }

    @Override
    public boolean isDone() {
        return futureResponse.isDone();
    }

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        try {
            return get(-1, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            throw new XxRpcException(e);
        }
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        try {
            // future get
            XxRpcResponse xxlRpcResponse = futureResponse.get(timeout, unit);
            if (xxlRpcResponse.getErrorMsg() != null) {
                throw new XxRpcException(xxlRpcResponse.getErrorMsg());
            }
            return xxlRpcResponse.getResult();
        } finally {
            stop();
        }
    }

    // ---------------------- 消费端线程异步响应 ----------------------

    private static ThreadLocal<XxRpcInvokeFuture> threadInvokerFuture = new ThreadLocal<XxRpcInvokeFuture>();

    /**
     * 获取异步响应
     *
     * @param type 类
     * @param <T> 泛型
     * @return 响应对象
     */
    public static <T> Future<T> getFuture(Class<T> type) {
        Future<T> future = (Future<T>) threadInvokerFuture.get();
        threadInvokerFuture.remove();
        return future;
    }

    /**
     * 设置异步响应值
     *
     * @param future 异步响应参数
     */
    public static void setFuture(XxRpcInvokeFuture future) {
        threadInvokerFuture.set(future);
    }

    /**
     * 移除异步响应值
     */
    public static void removeFuture() {
        threadInvokerFuture.remove();
    }
}
